package com.boilermaker.mq.mqserver.manager;

import com.boilermaker.mq.common.Consumer;
import com.boilermaker.mq.common.ConsumerEnv;
import com.boilermaker.mq.common.MqException;
import com.boilermaker.mq.mqserver.VirtualHost;
import com.boilermaker.mq.mqserver.core.MSGQueue;
import com.boilermaker.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/*
 * 消费消息
 */
@Slf4j
public class ConsumeManager {
    // 关联其所属的 VirtualHost，以获取其中的数据
    private VirtualHost parent;
    // 线程池用于处理消费者自定义的回调函数
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    // 该阻塞队列记录该 VirtualHost 中队列的消息发布情况（队列名）
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 该线程用于扫描 tokenQueue，获取队列的消息发布信息，找到相应队列取出消息，交由 workPool 处理
    private Thread scannerThread;

    public ConsumeManager(VirtualHost parent) {
        this.parent = parent;
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    String queueName = tokenQueue.take();
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) throw new MqException("队列不存在");
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 设为后台线程，因为该线程会持续运行，若为前台线程会导致程序无法终止
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    // 添加一个订阅者
    public void addConsumer(String consumeTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) throw new MqException("队列不存在, queueName=" + queueName);
        ConsumerEnv consumerEnv = new ConsumerEnv(consumeTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumer(consumerEnv);
            // 如果此时队列中尚有未消费的消息，则需要立即消费掉
            int count = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < count; i++) {
                consumeMessage(queue);
            }
        }
    }

    // 消费消息
    private void consumeMessage(MSGQueue queue) {
        // 选择一个消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) return;
        // 取出一条消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) return;
        // 在线程池中处理该消息
        workPool.submit(() -> {
            try {
                // 先将消息放入待确认队列
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 执行回调函数（将消息推送至客户端）
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),
                        message.getBasicProperties(), message.getBody());
                // 如果为自动应答，则执行后立即删除消息
                if (luckyDog.isAutoAck()) {
                    // 删硬盘
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 删内存（待确认队列和消息中心）
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    log.info("消息被成功消费, queueName=" + queue.getName());
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}
